package yredis

import (
	"context"
	"dgo/MyProj/PrjHttpUtils"
	"dgo/goutils/utils"
	"dgo/plg/plgbase"
	"fmt"
	"gitee.com/ymofen/gobase/golog"
	"gitee.com/ymofen/supervalue"
	"github.com/go-redis/redis/v8"
	"strings"
	"sync/atomic"
	"time"
)

/**
  该插件是基于redis实现的一个订阅中心
      当对订阅中心进行订阅通道时(所有通道除外) 会对ws进行订阅,
      当收到(topic)主题数据时,通过ex(订阅中心发布)
      当插件SendKVs, 会通过ws进行发送

  redisclt 会管理多个连接

*/

type PlgExRedis struct {
	send_err_n   uint32
	send_n       uint32
	push_fail_n  uint32
	recvpub_flag int32
	plgbase.PlgBase

	cachewks *utils.MultiGroupTask

	redis_sub_kv    utils.GSyncMapGroup
	redis_sub_start time.Time

	redis_addr      string
	redis_user      string
	redis_pass      string
	redis_pool_size int
	redis_ctx       context.Context
	data_ex         *utils.SubscribeCenter // 同步播发数据
	conn_n          int
	request_attach  int
	warnlst         *utils.SyncCycleList
	redisclt        *redis.Client
	redisclt_sub    *redis.PubSub
	monitor_send    *utils.MonitorSpeedSizeRec
	monitor_recv    *utils.MonitorSpeedSizeRec
	monitor_push    *utils.MonitorSpeedSizeRec
	last_warning    string
	last_warn_redis string
	last_warn_disp  string

	UseCache bool
	cachelst *utils.QueueTask

	last_recv          string
	last_recv_t        time.Time
	last_recv_workflag byte

	last_pub_t int64

	redis_push_history utils.GSyncMapGroup
}

func (this *PlgExRedis) GetPushTopic(t time.Duration) string {
	var sb utils.BytesBuilder
	this.redis_push_history.Range(func(key, value interface{}) bool {
		if time.Since(value.(time.Time)) < t {
			if sb.Len() > 0 {
				sb.AppendStr(",")
			}
			sb.Appendf("%v", key)
		}
		return true
	})
	return sb.String()
}

func (this *PlgExRedis) innerSend(topic string, data interface{}) error {
	if this.redisclt == nil {
		return fmt.Errorf("Redis连接尚未进行创建")
	}

	atomic.AddUint32(&this.send_n, 1)
	this.redis_push_history.Set(topic, time.Now())

	if buf, ok := data.([]byte); ok {
		_, err := this.redisclt.Publish(this.redis_ctx, topic, buf).Result()
		if err != nil {
			atomic.AddUint32(&this.send_err_n, 1)
			return err
		}
		this.last_pub_t = time.Now().Unix()
		this.monitor_send.DeltaBuf(buf)
	} else if str, ok := data.(string); ok {
		buf := []byte(str)
		_, err := this.redisclt.Publish(this.redis_ctx, topic, buf).Result()
		if err != nil {
			atomic.AddUint32(&this.send_err_n, 1)
			return err
		}
		this.last_pub_t = time.Now().Unix()
		this.monitor_send.DeltaBuf(buf)
	} else if json, ok := data.(*wrapper.SuperValue); ok {
		str := json.AsJSONStringEx(false, false)
		_, err := this.redisclt.Publish(this.redis_ctx, topic, str).Result()
		if err != nil {
			atomic.AddUint32(&this.send_err_n, 1)
			return err
		}
		this.last_pub_t = time.Now().Unix()
		this.monitor_send.Delta(int64(len(str)))
	} else {
		atomic.AddUint32(&this.send_err_n, 1)
		return fmt.Errorf("未知的数据类型:%v", data)
	}
	return nil
}

func (this *PlgExRedis) GetSpeedStatus() string {
	if this.cachelst == nil || this.ReleaseFlag == 1 {
		return ""
	}
	var sb utils.BytesBuilder
	if len(this.last_warning) > 0 {
		sb.Appendf("warn:%s\n", this.last_warning)
	}

	if len(this.last_warn_redis) > 0 {
		sb.Appendf("warn-redis:%s\n", this.last_warn_redis)
	}

	sb.Appendf("push:%s\n", this.monitor_push.Info())
	sb.Appendf(" send:%s, sendgroup:%d, send err:%d\n", this.monitor_send.Info(), this.GetPushTopicCount(time.Minute), this.send_err_n)
	sb.Appendf(" push topic:%s\n", this.GetPushTopic(time.Minute))
	sb.Appendf(" recv:%s, recvtopic:%d\n", this.monitor_recv.Info(), this.redis_sub_kv.Count())
	sb.Appendf(" redis-pool(idle/total): %d/%d\n", this.redisclt.PoolStats().IdleConns, this.redisclt.PoolStats().TotalConns)

	if len(this.last_warning) > 0 {
		sb.Appendf(" warn:%s\n", this.last_warning)
	}
	if len(this.last_warn_redis) > 0 {
		sb.Appendf(" warn-redis:%s\n", this.last_warn_redis)
	}
	if this.UseCache {
		sb.Appendf(" cache:%d, push fail:%d\n", this.cachelst.Count(), this.push_fail_n)
		sb.Appendf(" cache worker:%s\n", this.cachelst.StatusString())
	}
	return sb.String()
}

func (this *PlgExRedis) SendKVs(topic string, kv ...interface{}) error {
	var data interface{}
	l := len(kv)
	if l == 1 {
		data = kv[0]
		if buf, ok := data.([]byte); ok {
			this.monitor_push.DeltaBuf(buf)
		} else if str, ok := data.(string); ok {
			buf := []byte(str)
			data = buf
			this.monitor_push.DeltaBuf(buf)
		}
	} else {
		json := wrapper.NewSVObject()
		json.SetKeyValues(kv...)
		buf := json.AsJSONBytesEx(false, false)
		this.monitor_push.DeltaBuf(buf)
		data = buf
	}

	if this.UseCache {
		err := this.cachewks.PostTaskFunc(topic, func() {
			err := this.innerSend(topic, data)
			if err != nil {
				this.last_warn_redis = fmt.Sprintf("[%s][%s]推送数据异常:%s", utils.NowString(), topic, err.Error())
			}
		})
		if err != nil {
			atomic.AddUint32(&this.push_fail_n, 1)
			this.last_warning = fmt.Sprintf("[%s][%s]压入队列异常:%s", utils.NowString(), topic, err.Error())
		}
		return err

		//cachelst := this.cachelst
		//if cachelst != nil {
		//	err := cachelst.Push(topic, data)
		//	if err != nil {
		//		atomic.AddUint32(&this.push_fail_n, 1)
		//		this.last_warning = fmt.Sprintf("[%s][%s]压入队列异常:%s", utils.NowString(), topic, err.Error())
		//	}
		//	return err
		//}

	} else {
		err := this.innerSend(topic, data)
		if err != nil {
			//this.last_warning = fmt.Sprintf("[%s][%s]推送数据异常:%s", utils.NowString(), topic, err.Error())
			this.last_warn_redis = fmt.Sprintf("[%s][%s]推送数据异常:%s", utils.NowString(), topic, err.Error())
		}
		return err
	}
}

func (this *PlgExRedis) innerPub(topic string, data interface{}) error {
	clt := this.redisclt
	if clt == nil {
		atomic.AddUint32(&this.send_err_n, 1)
		return fmt.Errorf("尚未建立连接")
	}
	if data != nil {
		if buf, ok := utils.TryGetBuf(data); ok {
			_, err := clt.Publish(this.redis_ctx, topic, buf).Result()
			if err != nil {
				this.last_warn_redis = fmt.Sprintf("[%s]topic:%s, err:%s", utils.NowString(), topic, err.Error())
				atomic.AddUint32(&this.send_err_n, 1)
				return err
			}
			this.last_pub_t = time.Now().Unix()
			this.monitor_send.DeltaBuf(buf)
			return nil
		} else {
			atomic.AddUint32(&this.send_err_n, 1)
			return fmt.Errorf("未知的数据")
		}
	} else {
		atomic.AddUint32(&this.send_err_n, 1)
		return fmt.Errorf("数据为nil")
	}
}

func (this *PlgExRedis) Pub(topic string, max int, args ...interface{}) int {
	if len(args) < 0 {
		return -1
	}
	data := args[0]
	if this.UseCache {
		err := this.cachewks.PostTaskFunc(topic, func() {
			errinner := this.innerPub(topic, data)
			utils.TryCallBack(data, this, errinner)
		})
		if err != nil {
			utils.TryCallBack(data, this, err)
			return -1
		}
		return 1
	} else {
		err := this.innerPub(topic, data)
		utils.TryCallBack(data, this, err)
		if err != nil {
			return -1
		}
		return 1
	}
}

func (this *PlgExRedis) DispatchData(channelid string, sender interface{}, data interface{}, tag interface{}) (err error) {
	return this.SendKVs(channelid, data)
}

/*
**

	接收通道数据
*/
func (this *PlgExRedis) checkStartRecvTopic() {
	if !atomic.CompareAndSwapInt32(&this.recvpub_flag, 0, 1) {
		return
	}
	utils.GoFunc2(func() {
		var innerDispChannelMsg = func(msg *redis.Message) {

			this.last_recv_t = time.Now()
			this.last_recv_workflag = 1
			t0 := time.Now()
			buf := []byte(msg.Payload)
			defer func() {
				this.last_recv_workflag = 0
				ms := time.Since(t0).Milliseconds()
				if ms > 500 {
					this.last_warn_disp = fmt.Sprintf("[%s][%s]分发数据超时耗时:%d(ms), buf:%s", utils.NowString(), msg.Channel, ms, utils.BufToHexStr(buf, 0, ""))
				}
			}()
			this.last_recv = fmt.Sprintf("[%s][%s][%d]", utils.NowString(), msg.Channel, len(buf))
			this.monitor_recv.DeltaBuf(buf)
			this.data_ex.DispatchData(msg.Channel, nil, buf, nil)
		}

		for {
			pub := this.redisclt_sub
			if pub != nil {
				this.redis_sub_start = time.Now()
				for msg := range this.redisclt_sub.Channel() {
					innerDispChannelMsg(msg)
				}
			}
			if this.ReleaseFlag == 1 {
				break
			} else {
				time.Sleep(time.Second)
			}
		}
		this.recvpub_flag = 0
	})
}

func (this *PlgExRedis) ReqDis() {

	pub := this.redisclt_sub
	if pub != nil {
		pub.Close()
		this.redisclt_sub = nil
	}

	clt := this.redisclt
	if clt != nil {
		clt.Close()
		this.redisclt = nil
	}
}

/*
*
通道第一次创建的时候订阅
*/
func (this *PlgExRedis) innerSubscribeRedis(channelid string) bool {
	wlt := this.redisclt
	if wlt == nil {
		this.warnlst.Push(fmt.Sprintf("[%s]尚未分配redisclt", utils.NowString()))
		return false
	}
	var err error
	pub := this.redisclt_sub
	if pub == nil {
		this.redisclt_sub = wlt.Subscribe(this.redis_ctx, channelid)
		pub = this.redisclt_sub
		this.checkStartRecvTopic()
	} else {
		err = pub.Subscribe(this.redis_ctx, channelid)
	}

	if err != nil {
		this.warnlst.Push(fmt.Sprintf("[%s]订阅[%s]失败:%s", utils.NowString(), channelid, err.Error()))
		return false
	}
	this.redis_sub_kv.Set(channelid, time.Now())
	return true
}

func (this *PlgExRedis) innerUnSubscribeRedis(channelid string) {
	ok := this.redis_sub_kv.RemoveEx(channelid)
	if ok {
		pub := this.redisclt_sub
		if pub != nil {
			pub.Unsubscribe(this.redis_ctx, channelid)
		}
	}
}

func (this *PlgExRedis) SubscribeChannel(channelid string, id interface{}, cb utils.SubscribeCallback) {
	this.data_ex.SubscribeChannel(channelid, id, cb)

	e1 := utils.DefaultWorkers().PostTaskFunc(this, func() {
		if this.redis_sub_kv.Get(channelid, utils.ZeroTime).(time.Time).IsZero() {
			if this.innerSubscribeRedis(channelid) {
				golog.Infof("[%s]订阅成功", channelid)
				this.data_ex.DispatchData(channelid, id, nil, 1) // 订阅成功
			} else {
				this.data_ex.DispatchData(channelid, id, nil, 0) // 订阅失败
				golog.Warnf(fmt.Sprintf("[%s]订阅失败", channelid))
			}
		} else {
			this.data_ex.DispatchData(channelid, id, nil, 1) // 订阅成功
			// golog.Warnf( fmt.Sprintf("[%s]已经订阅", channelid))
		}
	})
	if e1 != nil {
		this.data_ex.DispatchData(channelid, id, nil, 0) // 订阅失败
		golog.Warnf(fmt.Sprintf("[%s]投递订阅信息失败:%s", channelid, e1.Error()))
	}
}

func (this *PlgExRedis) UnsubscribeChannel(channelid string, id interface{}) bool {
	return this.data_ex.UnsubscribeChannel(channelid, id)
}

func (this *PlgExRedis) GetSubscribeCount() int {
	return this.redis_sub_kv.Count()
}

func (this *PlgExRedis) GetPushTopicCount(t time.Duration) int {
	n := 0
	this.redis_push_history.Range(func(key, value interface{}) bool {
		if time.Since(value.(time.Time)) < t {
			n++
		}
		return true
	})
	return n
}

func (this *PlgExRedis) OnHttpStatus(rec *PlgHttpRec) (rval interface{}, errcode int, err error) {
	itype := rec.QryAsInt("type", 0)
	if itype == 99 {
		json := wrapper.NewSVObject()
		json.SetKeyValue("redis-ver", redis.Version())

		clt := this.redisclt
		//sub := this.redisclt_sub
		if clt != nil {
			json.SetKeyValue("redis-addr", clt.Options().Addr)
			json.SetKeyValue("redis-pool-status", fmt.Sprintf("idle/total :%d/%d", clt.PoolStats().IdleConns, clt.PoolStats().TotalConns))
		}
		//if sub != nil && this.redis_sub_kv.Count() < 10 {
		//	json.SetKeyValue("sub-info", fmt.Sprintf("[%s]%s", utils.DateTimeStringForShortShow(this.redis_sub_start), sub.String()))
		//}
		json.SetKeyValue("recv-pub-flag", this.recvpub_flag)
		if this.last_recv_workflag == 1 {
			dura := time.Since(this.last_recv_t)
			json.SetKeyValue("last-recv", fmt.Sprintf("%s, 分发耗时:%s", this.last_recv, utils.DurationStr(dura)))
		} else {
			json.SetKeyValue("last-recv", this.last_recv)
		}
		json.SetKeyValues("send-info", this.monitor_send.Info(),
			"recv-info", this.monitor_recv.Info(),
			"push-info", this.monitor_push.Info(),
			"send-err-n", this.send_err_n,
			"send-n", this.send_n,
			"last-warning", this.last_warning,
			"last-warn-redis", this.last_warn_redis,
			"last-warn-disp", this.last_warn_disp,
			"sub-n", this.redis_sub_kv.Count(),
		)
		topic_n := this.GetPushTopicCount(time.Minute)
		json.SetKeyValue("topic-n", fmt.Sprintf("recv %d, sendtopic-n %d(in one minutes)", this.redis_sub_kv.Count(), topic_n))
		//if topic_n < 10 {
		//	json.SetKeyValue("pushtopic", this.GetPushTopic(time.Minute))
		//}

		json.SetKeyValue("use-push-cache", this.UseCache)

		if this.UseCache {
			//json.SetKeyValue("cachelst", fmt.Sprintf("cache-n:%d, pushfail:%d", this.cachelst.Count(), this.push_fail_n, this.cachelst))
			if this.cachelst != nil {
				json.SetKeyValue("cache-lst", this.cachelst.StatusString())
			}
			if this.cachewks != nil {
				json.SetKeyValue("cache-wks", this.cachewks.StatusSimpleString())
			}
		}

		PrjHttpUtils.JSON2TableRec(rec.RespW, "", json.AsJSONString(false))
	} else if itype == 97 {
		if rec.QryAsInt("json", 0) == 1 {
			rec.ResponseJSONValues("last_request_t", this.last_pub_t, "last_response_t", this.last_recv_t.Unix(), "now", time.Now().Unix())
		} else {
			rec.ResponseText(fmt.Sprintf("%d,%d,%d", this.last_pub_t, this.last_recv_t.Unix(), time.Now().Unix()))
		}
	} else if itype == 1 {
		var sb utils.BytesBuilder
		searchval := rec.QryAsString("searchval", "")
		n := 0
		this.redis_sub_kv.Range(func(key, value interface{}) bool {
			n++
			channelid := key.(string)
			if len(searchval) == 0 || strings.Contains(channelid, searchval) {
				sb.Appendf("%d: [%s][start:%s]\r\n", n, channelid, utils.DateTimeString3(value.(time.Time)))
			}
			return true
		})
		rec.RespW.Write(sb.Bytes())
	} else if itype == 2 {
		rec.ResponseText(this.data_ex.StatusString())
	} else if itype == 3 {
		rec.ResponseSyncListN(this.warnlst, rec.QryAsString("searchval", ""), rec.QryAsInt("n", 0))
	} else if itype == 4 {
		utils.DefaultWorkers().PostTaskFunc(this, func() {
			this.ReqDis()
			this.innerCheckConnect()
			this.ReSub()
		})
		rec.ResponseText("ok")
	} else if itype == 5 {
		rec.ResponseText(this.GetSpeedStatus())
	} else if itype == 6 {
		sub := this.redisclt_sub
		if sub != nil {
			rec.ResponseText(sub.String())
		} else {
			rec.ResponseText("none")
		}
	} else if itype == 7 {
		var sb utils.BytesBuilder
		sb.Appendf("now:%s\r\n", utils.NowString())
		sb.Appendf("publst in one minutes\r\n")
		sb.Appendf("=======================================")
		sb.AppendStr(this.GetPushTopic(time.Minute))
		rec.ResponseText(sb.String())
		//rec.ResponseText(this.GetSpeedStatus())
	} else {
		var sb utils.BytesBuilder
		PrjHttpUtils.AppendIndexBody(&sb, "status", "status?type=99")

		PrjHttpUtils.AppendIndexBody(&sb, "subscribe-list", "status?type=1&searchval=")
		PrjHttpUtils.AppendIndexBody(&sb, "data_ex-status", "status?type=2")
		PrjHttpUtils.AppendIndexBody(&sb, "warn-list", "status?type=3&searchval=")
		PrjHttpUtils.AppendIndexBody(&sb, "reconnect->resub", "status?type=4")
		PrjHttpUtils.AppendIndexBody(&sb, "speedstatus", "status?type=5")
		PrjHttpUtils.AppendIndexBody(&sb, "sublst", "status?type=6")
		PrjHttpUtils.AppendIndexBody(&sb, "publst", "status?type=7")

		PrjHttpUtils.ResponseBodyHtml(rec.RespW, sb.String())
	}

	return
}

func (this *PlgExRedis) OnHttpDo(rec *PlgHttpRec) (rval interface{}, errcode int, err error) {
	action := rec.QryAsInt("action", 0)
	if action == 1 {

	} else {
		json := wrapper.NewSVObject()
		json.SetKeyValues("action-1", "请求断线")
		PrjHttpUtils.JSON2TableRec(rec.RespW, "", json.String())
	}
	return
}

/*
**

	{
	  "id":"yredis",
	  "type":"yredis",
	  "conf":{
		"addr":"119.96.169.117:6379",
	    "subscribe":"RAW-RTCM*",
	    "user":"",
	    "pass":"",
	    "pool-size":100
	   }
	}
*/
func (this *PlgExRedis) ReloadConf(conf *wrapper.SuperValue) error {
	if this.warnlst == nil {
		this.warnlst = utils.NewSyncCycleList()
		this.warnlst.ConfigMaxSize(1024)
	}
	if this.redis_ctx == nil {
		this.redis_ctx = context.Background()
		this.redis_pool_size = 50
	}

	if this.monitor_send == nil {
		this.monitor_send = utils.NewMonitorSpeedSizeRec()
		this.monitor_recv = utils.NewMonitorSpeedSizeRec()
		this.monitor_push = utils.NewMonitorSpeedSizeRec()
		utils.Monitor.AddMonitorRec(this.monitor_send, this.monitor_send)
		utils.Monitor.AddMonitorRec(this.monitor_recv, this.monitor_recv)
		utils.Monitor.AddMonitor(this.monitor_push)
	}

	if this.cachewks == nil {
		this.cachewks = utils.NewMultiGroupTask()
		this.cachewks.ConfigWorkType(2)
		this.cachewks.ConfigMinWork(3).ConfigMaxWork(5)
		this.cachewks.Start()
		utils.Monitor.AddMonitor(this.cachewks.Monitor_push)
	}

	isChanged := false

	conf.CheckExistsStrFunc("user", func(strv string) {
		strv = utils.ReplacePlaceholder(strv, "$$", "$$")
		if strv != this.redis_user {
			this.redis_user = strv
			isChanged = true
		}
	})

	conf.CheckExistsStrFunc("pass", func(strv string) {
		strv = utils.ReplacePlaceholder(strv, "$$", "$$")
		if strv != this.redis_pass {
			this.redis_pass = strv
			isChanged = true
		}
	})

	conf.CheckExistsStrFunc("addr", func(strv string) {
		strv = utils.ReplacePlaceholder(strv, "$$", "$$")
		strv = plgbase.ReplacePlaceholderForFileName(strv)
		if this.redis_addr != strv {
			this.redis_addr = strv
			isChanged = true
		}
	})

	conf.CheckExistsStrFunc("usecache", func(strv string) {
		v := utils.TryStrToBool(strv, false)
		if this.UseCache != v {
			//if v {
			//	if this.cachelst == nil {
			//		this.cachelst = utils.NewQueueTaskFunc()
			//		this.cachelst.ConfigMaxQueueSize(20480)
			//		this.cachelst.ConfigWorkNum(3, 5)
			//		this.cachelst.ConfigTaskCallBackFunc(func(args ...interface{}) {
			//			if len(args) == 2 {
			//				topic := args[0].(string)
			//				err := this.innerSend(topic, args[1])
			//				if err != nil {
			//					this.last_warn_redis = fmt.Sprintf("[%s][%s]推送数据异常:%s", utils.NowString(), topic, err.Error())
			//				}
			//			}
			//		})
			//		this.cachelst.Start()
			//		golog.Infof( "开启缓存推送模式!")
			//	}
			//}
			//this.UseCache = v
			//if !v {
			//	if this.cachelst != nil {
			//		this.cachelst.Stop()
			//		this.cachelst = nil
			//	}
			//}

			this.UseCache = v
		}
	})

	conf.CheckExistsStrFunc("pool-size", func(strv string) {
		n := utils.StrToIntDef(strv, 50)
		if n != this.redis_pool_size {
			this.redis_pool_size = n
			isChanged = true
			this.cachewks.ConfigMaxWork(int32(n))
		}
	})

	if isChanged {
		utils.DefaultWorkers().PostTaskFunc(this, func() {
			this.ReqDis()
			this.innerCheckConnect()
			this.ReSub()
		})
	}

	if this.data_ex == nil {
		this.data_ex = utils.NewSubscribeCenter()
		this.data_ex.OnUnSubChannel = func(channelid string, n int) {
			if n == 0 {
				if channelid == "*" {
					return
				}
				utils.DefaultWorkers().PostTaskFunc(this, func() {
					this.innerUnSubscribeRedis(channelid)
				})
			}
		}
	}

	return nil
}

func (this *PlgExRedis) innerCheckConnect() {
	if len(this.redis_addr) > 0 && this.redisclt == nil {
		this.redisclt = redis.NewClient(&redis.Options{
			Addr:     this.redis_addr,
			PoolSize: this.redis_pool_size,
			Username: this.redis_user,
			Password: this.redis_pass,
		})
		if this.redisclt_sub != nil {
			this.redisclt_sub.Close()
			this.redisclt_sub = nil
		}
		time.AfterFunc(time.Second, func() {
			cmd := this.redisclt.Ping(ctx_background)
			_, err := cmd.Result()
			if err != nil {
				this.last_warning = fmt.Sprintf("[%s]redis connect ping err:%s", utils.NowString(), err.Error())
				golog.Warnf("[%s]与Redis[%s]建立连接失败, PING err:%s", this.PlugId, this.redis_addr, err.Error())
			} else {
				golog.Infof("[%s]与Redis[%s]建立连接成功, PING成功", this.PlugId, this.redis_addr)
			}
		})

	}
}

func (this *PlgExRedis) ReSub() {
	this.redis_sub_kv.Range(func(key, value interface{}) bool {
		this.innerSubscribeRedis(key.(string))
		return true
	})
}

func (this *PlgExRedis) OnObjectNotify(sender interface{}, notifytype int, data interface{}) (rval interface{}, err error) {
	if notifytype == utils.NOTIFY_TYPE_SETCONF {
		if conf, ok := data.(*wrapper.SuperValue); ok {
			err = this.ReloadConf(conf)
		} else {
			err = fmt.Errorf("非法的JSON配置")
		}
	} else if notifytype == utils.NOTIFY_TYPE_RELEASE {
		this.ReleaseFlag = 1

		if this.monitor_send != nil {
			utils.Monitor.RemoveMonitorRec(this.monitor_send)
			utils.Monitor.RemoveMonitorRec(this.monitor_recv)
			utils.Monitor.RemoveMonitorRec(this.monitor_push)
		}

		if this.cachelst != nil {
			this.cachelst.Stop()
		}

		this.ReqDis()

		if this.redisclt != nil {
			this.redisclt.Close()
		}

		if this.cachewks != nil {
			utils.Monitor.RemoveMonitorRec(this.cachewks.Monitor_push)
			this.cachewks.Stop()
		}
	}
	return
}

func CreatePlgExRedisFunc(args ...interface{}) (rval interface{}, err error) {
	obj := &PlgExRedis{}
	if len(args) > 0 {
		if str, ok := args[0].(string); ok {
			obj.PlugId = str
		}
	}
	rval = obj
	return
}
